package cn.dfun.sample.flink.apitest

import java.util
import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.functions.{RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state._
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
  * 1.state示例
  * 2.使用state实现连续10秒温度增加报警
  * 3.checkpoint相关配置
  */
// 每个任务保存完后合并的合照为checkpoint
object StateTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 开发环境没必要配置checkpoint
    // 生产环境主要配置前三项
    // 触发checkpoint间隔时间,实际耗费时间要更长
    // 并不是所有任务保存checkpoint的时间
    env.enableCheckpointing(1000L)
    // 可优化
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
    // 超时时间,最大的时间耗费在远程传输
    // 异步checkpoint在内存中保存一个副本
    env.getCheckpointConfig.setCheckpointTimeout(60000L)
    // 最多允许出现的checkpoint,默认1
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
    // 两个checkpoint的最小间隔时间,该配置会覆盖MaxConcurrentCheckpoints
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // 是否优先使用checkpoint(另外有savepoint)
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
    // 允许checkpoint失败次数,0表示checkpoint失败则任务失败
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)
    // 重启策略 固定时间间隔重启 尝试重启次数|重启时间间隔
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))
    // 失败率
    env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(5, TimeUnit.MINUTES), Time.of(10, TimeUnit.MINUTES)))

    val inputStream = env.socketTextStream("node-01", 7777)

    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      .uid("1") // 推荐设置uid,便于恢复

    // 需求: 连续两个温度差在10度以上报警
    val alertStream = dataStream
        .keyBy(_.id)
          // 通过RichFunction实现
//        .flatMap(new TempChangeAlert(10.0))
        // 另一种实现思路
        .flatMapWithState[(String, Double, Double), Double]({
        // 初始状态
        case (data: SensorReading, None) => (List.empty, Some(data.temperature))
        case (data: SensorReading, lastTemp: Some[Double]) => {
          val diff = (data.temperature - lastTemp.get).abs
          if(diff > 10.0)
            (List((data.id, lastTemp.get, data.temperature)), Some(data.temperature))
          else
            (List.empty, Some(data.temperature))
        }})
    alertStream.print()
    env.execute("state test")
  }
}

class TempChangeAlert(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTempState", classOf[Double]))

  override def flatMap(in: SensorReading, collector: Collector[(String, Double, Double)]): Unit = {
    val lastTemp = lastTempState.value()
    val diff = (in.temperature - lastTemp).abs
    if(diff > threshold)
      collector.collect((in.id, lastTemp, in.temperature))
    // 更新当前温度
    lastTempState.update(in.temperature)
  }
}

// Keyed state 必须定义在RichFunction中
class MyRickFunc extends RichMapFunction[SensorReading, String] {
  var valueState :ValueState[Double] = _
  // 命名必须唯一
  lazy val listState: ListState[Int] = getRuntimeContext.getListState(new ListStateDescriptor[Int]("liststate", classOf[Int]))
  lazy val mapState: MapState[String, Double] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Double]("mapstate", classOf[String], classOf[Double]))
  lazy val reducingState: ReducingState[SensorReading] = getRuntimeContext.getReducingState(
    new ReducingStateDescriptor[SensorReading]("reducingstate", new MyReducer, classOf[SensorReading]))

  override def open(parameters: Configuration): Unit = {
    valueState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("valuestate", classOf[Double]))
  }
  override def map(in: SensorReading): String = {
    // 读取state
    val myV = valueState.value()
    // 修改数据
    valueState.update(in.temperature)

    listState.add(1)
    val list = new util.ArrayList[Int]();
    list.add(2)
    list.add(3)
    listState.addAll(list)
    // 获取list值,可遍历
    listState.get()

    // mapState操作
    mapState.contains("sensor_1")
    mapState.get("sensor_1")
    mapState.put("sensor_1", 1.2)

    reducingState.get()
    // 聚合
    reducingState.add(in)
    in.id
  }
}
